Flask 应用之搭建 API 服务的笔记

Flask 是基于 Werkzeug 和 Jinja2 的 Python 轻量级 Web 应用框架，适用于轻量级的服务，例如: 简单网页开发以及搭建 RESTful API 服务。主要整理了搭建 API 过程中遇到的一些问题，以及相关流程中的代码。

通用 API 请求

直接通过链接不带任何请求数据以及查询参数的请求，这类请求下可以直接搭建相关相应的网页返回数据即可：

example.py >unfolded

import flask
from flask import request, jsonify

app = flask.Flask(__name__)
app.config["DEBUG"] = True

# Create some test data for our catalog in the form of a list of dictionaries.
books = [
    {'id': 0,
     'title': 'A Fire Upon the Deep',
     'author': 'Vernor Vinge',
     'first_sentence': 'The coldsleep itself was dreamless.',
     'year_published': '1992'},
    {'id': 1,
     'title': 'The Ones Who Walk Away From Omelas',
     'author': 'Ursula K. Le Guin',
     'first_sentence': 'With a clamor of bells that set the swallows soaring, the Festival of Summer came to the city Omelas, bright-towered by the sea.',
     'published': '1973'},
    {'id': 2,
     'title': 'Dhalgren',
     'author': 'Samuel R. Delany',
     'first_sentence': 'to wound the autumnal city.',
     'published': '1975'}
]

@app.route('/api/v1/resources/books/all', methods=['GET'])
def api_all():
    return jsonify(books)

上面提供的 API 提供了 GET 方法，直接通过 http://<host>/api/v1/resources/books/all 链接可以得到所有数据 JSON 结果。

带参数查询的 API 请求

带参数的请求包括通过在链接中添加查询参数访问或者通过表单方式请求，前者的参数解析需要通过 request 的 args 属性得到相关的，查询的单一参数在链接中一般是以 ?<key>=<value> 形式跟在链接后。如果需要查询多个参数时，是以 ?<key1>=<value1>&<key2>=<value2> 形式；后者的参数解析是通过解析表单数据，键值是以请求中 body 参数传入。下面的示例是需要带访问参数的请求查询:

example1.py

# 省略了部分初始化 app 代码
from flask import request, jsonify

@app.route('/api/v1/resources/books', methods=['GET'])
def api_id():
    # Check if an ID was provided as part of the URL.
    # If ID is provided, assign it to a variable.
    # If no ID is provided, display an error in the browser.
    if 'id' in request.args:
        id = int(request.args['id'])
    else:
        return "Error: No id field provided. Please specify an id."

    # Create an empty list for our results
    results = []

    # Loop through the data and match results that fit the requested ID.
    # IDs are unique, but other fields might return many results
    for book in books:
        if book['id'] == id:
            results.append(book)

    # Use the jsonify function from Flask to convert our list of
    # Python dictionaries to the JSON format.
    return jsonify(results)

以上的 API 可以通过链接 http://<host>/api/v1/resources/books?id=1 查询 id 为 1 的书籍。

定制化响应异常

在 API 服务中异常处理包括了本身代码异常处理，以及提供服务的异常处理。服务的异常包括请求格式不符合要求、连接异常等，某些异常有专门的响应代码表达响应的结果。为了规范响应，使用 errorhandler 可以通过定制化响应的响应结果：

example3.py >folded

# 省略了部分初始化 app 代码
from flask import request, jsonify
import sqlite3

def dict_factory(cursor, row):
    d = {}
    for idx, col in enumerate(cursor.description):
        d[col[0]] = row[idx]
    return d


@app.route('/', methods=['GET'])
def home():
    return '''<h1>Distant Reading Archive</h1>
<p>A prototype API for distant reading of science fiction novels.</p>'''

# 404 异常
@app.errorhandler(404)
def page_not_found(e):
    return "<h1>404</h1><p>The resource could not be found.</p>", 404

@app.route('/api/v1/resources/books', methods=['GET'])
def api_filter():
    query_parameters = request.args

    id = query_parameters.get('id')
    published = query_parameters.get('published')
    author = query_parameters.get('author')

    query = "SELECT * FROM books WHERE"
    to_filter = []

    if id:
        query += ' id=? AND'
        to_filter.append(id)
    if published:
        query += ' published=? AND'
        to_filter.append(published)
    if author:
        query += ' author=? AND'
        to_filter.append(author)
    if not (id or published or author):
	# 未找到信息时，返回指定的 404 信息
        return page_not_found(404)

    query = query[:-4] + ';'

    conn = sqlite3.connect('books.db')
    conn.row_factory = dict_factory
    cur = conn.cursor()

    results = cur.execute(query, to_filter).fetchall()

    return jsonify(results)

上面代码定制化处理了 404 响应结果

开发与测试 SQLAlchemy

SQLAlchemy 是作为 Flask 扩展工具，其主要主要作用是处理数据库数据。在开发或者测试阶段进行交互环境下操作时，需要调用应用上下文的 push 方法才能进行进行后续的数据操作:

app.py >unfolded

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

def create_app():
    app = Flask(__name__)
    db.init_app(app)
    return app

上面的代码，包括了初始化一个应用实例以及初始化应用数据库（通过 db.init_app 方法可以初始化应用的需要的数据库）。在交互时需要以下面的方式使用 push 方法:

>unfolded

>>> from app import create_app
>>> app = create_app()
>>> app.app_context().push()

外链图片显示

Flask 中直接使用 URL 的外链，可能会存在无法展示的问题。解决这个问题，需要将链接转换为具体的数据值，例如将数据直接数据读取保存为 base64 编码数据即可以解决：

1. 需要将数据转换为 base64 编码数据

   example3.py >unfolded

    # 代码部分省略了数据保存部分
   import requests
   import base64
   
   # 如果是需要保存在数据库中，可以不需要使用解码，即保留字节型数据（例如 MySQL 中保存为 BLOB 数据，保存的数据格式应该是子节型数据，不应该使用解码）
   url = “https://img1.doubanio.com/view/photo/s_ratio_poster/public/p2196652087.jpg”
   content = base64.b64encode(requests.get(url).content)

2. 在渲染阶段需要将图片的值解码后，赋值给 src 属性。下面的代码包括了两个部分，一个是从数据库查询数据并解码，另一个是利用 Jinja2 的模版格式化图片

   app.py >unfolded

   # 代码部分省略了 app 初始化等过程
   @app.route("/hello")
   def index():
       series = Series.query.limit(10)
       for item in series:
           item.cover = item.cover.decode("ascii")
       
       return render_template("index.html", series=series)

3. 模版格式化图片

   index.html >unfolded

   # 代码部分省略了页面其他部分

参考

1. Creating Web APIs with Python and Flask
2. Flask 文档